package cn.tedu.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class SyncConfirm {
    public static void main(String[] args) throws Exception{
        //1.配置连接信息
        ConnectionFactory cf = new ConnectionFactory();
        cf.setHost("127.0.0.1");
        cf.setPort(5672);//5672是默认端口，可以省略
        cf.setUsername("guest");
        cf.setPassword("guest");
        //2.创建tcp连接对象
        Connection nc = cf.newConnection();
        //3.创建channel
        Channel cc = nc.createChannel();
        //4.创建一个随机队列
        String queueName = cc.queueDeclare().getQueue();
        //5.开启发布确认
        cc.confirmSelect();
        /*
          存储未确认的消息
          ConcurrentSkipListMap线程安全有序的一个hash表，适用于高并发情况
          1.将序号和消息进行关联
          2.支持并发访问
          3.只要给序列号，轻松批量删除
         */
        ConcurrentSkipListMap<Long,String> map = new ConcurrentSkipListMap<>();
        /*
          监听器的回调，确认收到一个回调
            1.消息的序列号
            2.true 可以确认小于或等于该序列号的消息  false就确认当前序列号的消息
         */
        ConfirmCallback ackCallback = new ConfirmCallback() {
            @Override
            public void handle(long l, boolean b) throws IOException {
                System.out.println("发送消息成功的序列号:"+l);
                //如果是true，返回的是小于或等于该序列号的未确认的消息
                if(b){
                    //传入true，确认小于或等于当前序列号的所有消息
                    ConcurrentNavigableMap<Long, String> confirmed
                            = map.headMap(l, true);
                    confirmed.clear();
                }else{
                    map.remove(l);
                }
            }
        };
        //发布失败的回调
        ConfirmCallback unAckCallback = new ConfirmCallback() {
            @Override
            public void handle(long l, boolean b) throws IOException {
                System.out.println("消息发布失败");
            }
        };
        long start = System.currentTimeMillis();
        for (int i = 1; i <= 1000; i++) {
            String msg = "消息："+i;
            long nextPublishSeqNo = cc.getNextPublishSeqNo();//可以获取到下一个消息的序列号
            //通过序列号与消息进行关联，全部都是未确认的消息
            map.put(nextPublishSeqNo,msg);
            cc.basicPublish("",queueName,null,msg.getBytes());
        }
        long end = System.currentTimeMillis();
        System.out.println("发送1000条消息数据使用的时间："+(end-start));
    }
}
